package com.zyf.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Malegod_xiaofei
 * @create 2024-08-18-22:39
 */
public class MySink extends AbstractSink implements Configurable {

    // 声明数据的前后缀
    private String prefix;
    private String subfix;

    // 创建 Logger 对象
    private Logger logger = LoggerFactory.getLogger(MySink.class);

    @Override
    public void configure(Context context) {

        prefix = context.getString("pre", "pre-");
        subfix = context.getString("sub");
    }

    @Override
    public Status process() throws EventDeliveryException {

        // 1.获取 Channel 并开启事务
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        transaction.begin();

        // 2.从 Channel 中抓取数据打印到控制台
        try {
            // 2.1 抓取数据
            Event event;
            while (true) {
                event = channel.take();
                if (event != null) {
                    break;
                }
            }
            // 2.2 处理数据
            logger.info(prefix + new String(event.getBody()) + subfix);

            // 2.3 提交事务
            transaction.commit();
            return Status.READY;
        } catch (Exception e) {
            e.printStackTrace();
            // 2.4 回滚事务
            transaction.rollback();
            return Status.BACKOFF;
        } finally {
            transaction.close();
        }
    }
}